package com.example.rxjava2;

import android.content.Intent;
import android.annotation.SuppressLint;
import android.os.Handler;
import android.os.Message;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.ImageView;
import android.widget.TextView;

import com.example.rxjava2.utils.RxTimerUtil;


import com.example.rxjava2.activity.ThreadActivity;

import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import retrofit2.Response;

@SuppressLint("CheckResult")
public class MainActivity extends AppCompatActivity implements View.OnClickListener {
    private static final String TAG = MainActivity.class.getSimpleName();
    private TextView result;
    private Disposable disposable;
    private int i = 0;
    private ImageView imageview1;
    private ImageView imageview2;
    private ImageView imageview3;
    @SuppressLint("HandlerLeak")
    private Handler mHandler = new Handler() {
        @Override
        public void handleMessage(Message msg) {
            super.handleMessage(msg);
            int what = msg.what;
            switch (what) {
                case 1001:
                    Log.i(TAG, "handleMessage: " + "get message ");
                    break;

                default:

                    break;
            }

        }
    };

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        result = findViewById(R.id.result);
        imageview1 = findViewById(R.id.imageview1);
        imageview2 = findViewById(R.id.imageview2);
        imageview3 = findViewById(R.id.imageview3);

        findViewById(R.id.btn1).setOnClickListener(this);
        findViewById(R.id.btn2).setOnClickListener(this);
        findViewById(R.id.btn3).setOnClickListener(this);
        findViewById(R.id.flatmap).setOnClickListener(this);
        findViewById(R.id.concatmap).setOnClickListener(this);
        findViewById(R.id.onnext).setOnClickListener(this);
        findViewById(R.id.filter).setOnClickListener(this);
        findViewById(R.id.skip).setOnClickListener(this);
        findViewById(R.id.take).setOnClickListener(this);
        findViewById(R.id.timer).setOnClickListener(this);
        findViewById(R.id.intravel).setOnClickListener(this);
        findViewById(R.id.concat).setOnClickListener(this);
        findViewById(R.id.distinct).setOnClickListener(this);
        findViewById(R.id.buffer).setOnClickListener(this);
        findViewById(R.id.publish).setOnClickListener(this);
        findViewById(R.id.test1).setOnClickListener(this);
        findViewById(R.id.test2).setOnClickListener(this);
        findViewById(R.id.btn_timer).setOnClickListener(this);
        findViewById(R.id.btn_rx_timer).setOnClickListener(this);
        findViewById(R.id.stop).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                disposable.dispose();
            }
        });

    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.btn1:
                testCreat();
                break;

            case R.id.btn2:
//                 "合并事件专用," +
//                        "分别从两个上游事件中各取出一个组合," +
//                        "一个事件只能被使用一次，顺序严格按照事件发送的顺序," +
//                        "最终下游事件收到的是和上游事件最少的数目相同（必须两两配对，多余的舍弃)";
                testZip();
            case R.id.btn3:
//                * 基本是RxJava 最简单的操作符了
//                * 作用是对上游发送的每一个事件应用一个函数，使得每一个事件都按照指定的函数去变化
                testMap();
                break;
            case R.id.flatmap:
                /*
                 * * FlatMap将一个发送事件的上游Observable变换成多个发送事件的Observables，
                 * 然后将它们发射的时间合并后放进一个单独的Observable里
                 * <p>
                 * [需要注意]flatMap并不保证事件的顺序
                 * */
                testFlatmap();
                break;
            case R.id.concatmap:
//                "concatMap作用和flatMap几乎一模一样，唯一的区别是它能保证事件的顺序"
                testConcatmap();
                break;
            case R.id.onnext:
                testOnntext();//onNext or onAfter
                break;
            case R.id.filter:
                //过滤 取出正确值
                testFilter();
                break;
            case R.id.skip:
                //接受一个long型参数，代表跳过多少个数目的事件再开始接收
                testSkip();
                break;
            case R.id.take:
                //用于指定订阅者最多收到多少数据
                testTake();
                break;
            case R.id.timer:
                testTimer();
                break;

            case R.id.intravel:
                testIntravel();
                break;

            case R.id.concat:
                textConcat();
                break;
            case R.id.distinct:
                //去重
                textDistinct();
                break;

            case R.id.buffer:
                /*
                 *buffer(count, skip)` 从定义就差不多能看出作用了，
                 *将 observable 中的数据按 skip（步长）分成最长不超过 count 的 buffer，然后生成一个 observable
                 **/
                textBuffer();
                break;

            case R.id.publish:
                testPublish();
                break;
            case R.id.test2:
                startActivity(new Intent(MainActivity.this,ThreadActivity.class));
                break;
            case R.id.btn_timer:
                Timer timer = new Timer();
                TimerTask timerTask = new TimerTask() {
                    @Override
                    public void run() {
                        Message message = new Message();
                        message.what = 1001;
                        mHandler.sendMessage(message);
                    }
                };

                timer.schedule(timerTask, 0, 2000);
                break;

            case R.id.btn_rx_timer:
                RxTimerUtil.interval(2000, new RxTimerUtil.IRxNext() {
                    @Override
                    public void doNext(long number) {
                        Log.i(TAG, "doNext: " + "number " + number);
                    }
                });
                break;
        }
    }



    private void useCreat() {
        Observable<News> observable = RetrofitUtil.getRestClient()
                .getNews("861a794733dd2b057e415d822bef9586", 5);

        observable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<News>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(News news) {
                        Log.i(TAG, "onNext: ");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: ");
                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "onComplete: ");
                    }
                });


//        news.subscribeOn(Schedulers.io())
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(new Observer<News>() {
//                    @Override
//                    public void onSubscribe(Disposable d) {
//
//                    }
//
//                    @Override
//                    public void onNext(News news) {
//                        Log.i(TAG, "onNext: ");
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        Log.i(TAG, "onError: ");
//                    }
//
//                    @Override
//                    public void onComplete() {
//                        Log.i(TAG, "onComplete: ");
//                    }
//                });

    }

    private void testPublish() {
        PublishSubject<Integer> publishSubject = PublishSubject.create();
        publishSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                result.append("------>" + integer + "\n");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

        publishSubject.onNext(1);
        publishSubject.onNext(2);

        publishSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                result.append("item :------->" + integer + "\n");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                result.append("complete 2");
            }
        });
        publishSubject.onNext(11);
        publishSubject.onNext(22);
        publishSubject.onComplete();

    }


    private void textBuffer() {
        Observable.just(1, 2, 3, 4, 5, 6, 7)
                .buffer(3, 2)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        result.append("integer  :" + integers + "\n");
                    }
                });
    }

    private void textDistinct() {
        Observable.just(1, 2, 3, 4, 5, 4, 3, 2)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        result.append("item :" + integer + "\n");
                    }
                });
    }

    private void textConcat() {
        Observable.concat(Observable.just(1, 2, 3), Observable.just(11, 22, 33))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        result.append(integer + "=----->" + "\n");
                    }
                });
    }

    private void testIntravel() {

        disposable = Observable.interval(3, 2, TimeUnit.SECONDS)
                .take(3)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        i++;
                        if (i % 3 == 0) {
                            imageview1.setImageResource(R.mipmap.bg_girl);
                        } else if (i % 3 == 1) {
                            imageview1.setImageResource(R.mipmap.bg_monkey);
                        } else if (i % 3 == 2) {
                            imageview1.setImageResource(R.mipmap.bg_monkey_king);
                        }
                        result.append(" along --->" + i + Thread.currentThread().getName() + "\n");
                    }
                });
    }

    private void testTimer() {
        Observable.timer(2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {

                        result.append(" == = = == =>" + aLong + Thread.currentThread().getName());
                    }
                });
    }

    private void testTake() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
                .take(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        result.append("take --->" + integer + "\n");
                    }
                });

    }

    private void testSkip() {
        Observable.just(1, 2, 3, 4, 5, 6)
                .skip(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        result.append(integer + "\n");
                    }
                });
    }

    private void testFilter() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer % 2 == 0;
                    }
                }).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                if (integer % 3 == 0) {
                    result.append(" %3----<" + "\n");
                }
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                result.append(integer + "\n");
            }
        });
    }

    private void testOnntext() {
        Observable.just(1, 2, 3, 4, 5).doOnNext(new Consumer<Integer>() {
            //        Observable.just(1, 2, 3, 4, 5).doAfterNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                if (integer % 2 == 0) {
                    result.append("=======》》》》" + "\n");
                }
            }

        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                result.append(" onnetxt --------->" + integer + "\n");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void testConcatmap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                ArrayList<String> strings = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    strings.add("item ::" + i);
                }
                return Observable.fromIterable(strings).delay(1000, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        result.append(s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        result.append("complete");
                    }
                });
    }

    private void testFlatmap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                int delayTime = (int) (1 + Math.random() * 10);
                return Observable.fromIterable(list).delay(1000, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        result.append(s + "\n");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        result.append("complete");
                    }
                });
    }

    private void testMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(4);
                e.onNext(5);
                e.onNext(7);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "This is result " + integer;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                result.append("onNext : " + s + "\n");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                result.append("complete");
            }
        });
    }

    private void testZip() {
        Observable.zip(getStringObservable(), getIntegerObservable(), new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Exception {
                return s + integer;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                result.append(s + "\n");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                result.append("complete:");
                Log.i(TAG, "onComplete: ");
            }
        });
    }

    private ObservableSource<Integer> getIntegerObservable() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                if (!e.isDisposed()) {
                    e.onNext(5);
                    e.onNext(4);
                    e.onNext(3);
                    e.onNext(2);
//                    e.onNext(1);
                }
            }
        });

    }

    private ObservableSource<String> getStringObservable() {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                if (!e.isDisposed()) {
                    e.onNext("item : --->1");
                    e.onNext("item : --->2");
                    e.onNext("item : --->3");
                    e.onNext("item : --->4");
                    e.onNext("item : --->5");
                }
            }
        });
    }

    private void testCreat() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: 0" + integer);
                if (integer == 3) {
                    disposable.dispose();
                }
                result.append(integer + "\n");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: " + e.toString());
            }

            @Override
            public void onComplete() {
                Log.i(TAG, "onComplete: initData");
            }
        });
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
//        if (disposable != null || !disposable.isDisposed()) {
//            disposable.dispose();
//        }
    }
}
